package org.zbus.proxy.tc;
  
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.zbus.broker.Broker;
import org.zbus.broker.BrokerConfig;
import org.zbus.broker.SingleBroker;
import org.zbus.kit.ConfigKit;
import org.zbus.rpc.RpcException;
import org.zbus.rpc.RpcProcessor;
import org.zbus.rpc.mq.Service;
import org.zbus.rpc.mq.ServiceConfig;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.eno.ENOInterface.ENOMsg;
import com.eno.InfoAgent.TCRSField;
import com.eno.etcpool.ENOProxy;
import com.eno.tcrs.ENORecordset;

class TcServiceImpl implements TcService{ 
	public static String toParamString(Map<String, Object> paras) {
		if (paras == null)
			return "";
		StringBuilder sb = new StringBuilder();
		int i = 0;
		for (Map.Entry<String, Object> entry : paras.entrySet()) {
			String value = (String)entry.getValue();
			if (value != null) {
				if (i > 0) {
					sb.append("&");
				}
				sb.append(entry.getKey()).append("=").append(value);
				i++;
			}
		}
		return sb.toString();
	}
	
	@Override
	public Object callString(int serviceId, int mainFuncId,
			int subFuncId, String paramString) { 
		try {
			ENOMsg msg = new ENOMsg(); 
			byte[] data = ENOProxy.getData(serviceId, mainFuncId, subFuncId,
					paramString.getBytes(), "", true, msg);
			if(data == null){
				throw new RpcException(msg.getMessage());
			}
		
			ENORecordset[] mrs = ENORecordset.constructMR(data, 0);

			if (mrs == null || mrs.length == 0) {
				throw new RpcException(msg.getMessage());
			}
 
			if (mrs[0].isErrorInfo()) { //error
				mrs[0].fetchFirst();
				throw new RpcException(mrs[0].getString(0));
			}
			JSONArray rsArrayJson = new JSONArray();
			for(int rsIdx=0; rsIdx<mrs.length; rsIdx++){ 
				ENORecordset rs = mrs[rsIdx];
				int cols = rs.fieldCount();
				rs.fetchFirst();   
				JSONArray rsJson = new JSONArray();
				while (!rs.isEOF()) { 
					JSONObject row = new JSONObject();
					for (int i = 0; i < cols; i++) {
						TCRSField field = rs.fieldDesc(i);
						String key = field.fieldName;
						String val = rs.toString(i); //damned ENO's bugs and fucking design
						if(val == null){
							val = "";
						}
						
						row.put(key, val); 
					} 
					rsJson.add(row); 
					rs.fetchNext();
				} 
				rsArrayJson.add(rsJson);
			} 
			
			return rsArrayJson;

		} catch (Exception e) { 
			throw new RpcException(e.getMessage(), e); 
		}  
	}
	
	@Override
	public Object callMap(int serviceId, int mainFuncId,
			int subFuncId, Map<String, Object> params) { 
		String paramString = toParamString(params); 
		return callString(serviceId, mainFuncId, subFuncId, paramString);
	}
	
}

public class TcProxy {  
	public static void main(String[] args) throws Exception{   
		String address = ConfigKit.option(args, "-b", "127.0.0.1:15555"); 
		int consumerCount = ConfigKit.option(args, "-c", 20);
		String service = ConfigKit.option(args, "-s", "ETC-172.24.181.70");
		
		RpcProcessor processor = new RpcProcessor();
		// 增加模块，模块名在调用时需要指定
		processor.addModule(new TcServiceImpl());
		
		String[] blocks = address.split("[;]");
		List<Broker> brokers = new ArrayList<Broker>();
		for(String serverAddress : blocks){
			serverAddress = serverAddress.trim();
			if(serverAddress.equals("")) continue; 
			
			//配置Broker
			BrokerConfig brokerCfg = new BrokerConfig();
			brokerCfg.setServerAddress(serverAddress); 
			Broker broker = new SingleBroker(brokerCfg);
			brokers.add(broker);
		}
		
		ServiceConfig config = new ServiceConfig(); 
		config.setMq(service);
		config.setConsumerCount(consumerCount);
		config.setBrokers(brokers.toArray(new Broker[0]));
		config.setMessageProcessor(processor);
	  
		@SuppressWarnings("resource")
		Service svc = new Service(config);
		svc.start();   
	} 
}
